package com.hilinos.im.core;

import cn.hutool.core.collection.ConcurrentHashSet;
import com.hilinos.common.common.ReturnCode;
import com.hilinos.common.util.AssertUtils;
import com.hilinos.common.util.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.ObjectUtils;

import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.Consumer;

/**
 * @author qhl
 */
@Slf4j
public abstract class ConnectorUtils<T> {

    private final Object LOCK = new Object();
    private final ThreadPoolExecutor executor_keep_alive = ThreadPoolUtils
            .createThreadPool("ConnectorUtils-keep-alive-", 1, 1, 0, 100);

    private final ThreadPoolExecutor executor = ThreadPoolUtils.createThreadPool("ConnectorUtils-Executor-", 5, 10, 0,
            10000);

    /**
     * 用户信息
     */
    private final Set<T> CONNECTORS = new ConcurrentHashSet<>();

    /**
     * 用户信息
     */
    private final Map<String, Set<T>> USER_CONNECTOR = new ConcurrentHashMap<>();

    public ConnectorUtils() {
        executor_keep_alive.execute(() -> {
            while (true) {
                try {
                    Thread.sleep(1000);
                } catch (Exception ignored) {
                }
                try {
                    foreach(connector -> {
                        if (needCloseConnector(connector)) {
                            delConnector(connector);
                            closeConnector(connector);
                        }
                    });
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 获取用户id
     *
     * @param connector
     * @return
     */
    public abstract String getUseridByConnector(T connector);

    /**
     * 关闭通道
     *
     * @param connector
     * @return
     */
    public abstract void closeConnector(T connector);

    /**
     * 是否需要关闭通道
     *
     * @param connector
     * @return
     */
    public abstract boolean needCloseConnector(T connector);

    /**
     * 是否需要关闭通道
     *
     * @param connector
     * @return
     */
    public abstract void sendMsg(T connector, String msg);

    /**
     * 添加连接
     *
     * @param connector
     */
    public void addConnect(T connector) {
        synchronized (LOCK) {
            AssertUtils.isNotEmpty(connector, ReturnCode.NORMAL_SYSTEM_ERROR);
            CONNECTORS.add(connector);
            String userId = getUseridByConnector(connector);
            if (!ObjectUtils.isEmpty(userId)) {
                Set<T> users = USER_CONNECTOR.get(userId);
                if (users == null) {
                    users = new HashSet<>();
                    users.add(connector);
                    USER_CONNECTOR.put(userId, users);
                }
                users.add(connector);
            }
        }
    }

    /**
     * 获取用户频道
     *
     * @param userId
     * @return
     */
    public Set<T> getUserConnectors(String userId) {
        synchronized (LOCK) {
            return USER_CONNECTOR.get(userId);
        }
    }

    /**
     * 断开连接
     *
     * @param connector
     * @return
     */
    public void delConnector(T connector) {
        synchronized (LOCK) {
            CONNECTORS.remove(connector);

            String userId = getUseridByConnector(connector);
            if (!ObjectUtils.isEmpty(userId)) {
                Set<T> connectors = USER_CONNECTOR.get(userId);
                if (ObjectUtils.isEmpty(connectors)) {
                    USER_CONNECTOR.remove(userId);
                } else {
                    connectors.remove(connector);
                    if (ObjectUtils.isEmpty(connectors)) {
                        USER_CONNECTOR.remove(userId);
                    }
                }
            }
        }
    }

    public void foreach(Consumer<T> consumer) {
        synchronized (LOCK) {
            CONNECTORS.forEach(t -> executor.execute(() -> consumer.accept(t)));
        }
    }

}
